package net.bwie.jtp.dim.trade.sink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class DimHbaseSinkFunction extends RichSinkFunction<String> {

    /**
     * 定义变量：数据库连接
     */
    private Connection connection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // s1-创建配置对象，设置属性值，比如连接ZK集群信息
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
        conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
        // s2-实例化Connection连接
        connection = ConnectionFactory.createConnection(conf);

    }
    @Override
    public void invoke(String value, Context context) throws Exception {

        JSONObject jsonObject = JSON.parseObject(value);
        String tableName = jsonObject.getString("hbase_table_name");
        String family = jsonObject.getString("family_name");
        String rowKey = jsonObject.getString("row_key_column");
        String operatetype = jsonObject.getString("operate_type");
        JSONObject dataObject = jsonObject.getJSONObject("operate_date");
        createHbaseTable(tableName,family);
//        {"operate_type":"insert","db_name":"jtp_mall","operate_date":{"taken_count":0,"start_time":"2025-02-10 01:01:01","create_time":"2025-01-11 12:37:05","condition_amount":99.00,"id":1,"range_type":"3301","coupon_name":"口红品类券","coupon_type":"3201","benefit_amount":30.00,"limit_num":100,"operate_time":"2025-01-11 12:37:05"},"hbase_table_name":"dim_coupon_info","table_name":"coupon_info","family_name":"info","ts_ms":1748918812638,"row_key_column":"id"}
//        dim> {"operate_type":"insert","db_name":"jtp_mall","operate_date":{"taken_count":0,"start_time":"2025-02-10 01:01:01","create_time":"2025-01-11 12:37:05","condition_amount":99.00,"id":1,"range_type":"3301","coupon_name":"口红品类券","coupon_type":"3201","benefit_amount":30.00,"limit_num":100,"operate_time":"2025-01-11 12:37:05"},"hbase_table_name":"dim_coupon_info","table_name":"coupon_info","family_name":"info","ts_ms":1748918812638,"row_key_column":"id"}

        Table table = connection.getTable(TableName.valueOf(tableName));
        if("insert".equals(operatetype)||"update".equals(operatetype)) {
            String rowKeyValue = dataObject.getString(rowKey);
            Put put = new Put(Bytes.toBytes(rowKeyValue));
            for (String columnName : dataObject.keySet()) {
                String columnValue = dataObject.getString(columnName);
                // 当列值不为null时，插入hbase表
                if(null != columnValue){
                    put.addColumn(
                            Bytes.toBytes(family), Bytes.toBytes(columnName) , Bytes.toBytes(columnValue)
                    );
                }
            }
            table.put(put);
        } else if ("delete".equals(operatetype)) {
            String rowKeyValue = dataObject.getString(rowKey);
            Delete delete = new Delete(Bytes.toBytes(rowKeyValue));
            table.delete(delete);
        }else {
            System.out.println("操作类型错误！");
        }
        table.close();
    }
    private void createHbaseTable(String table, String family) throws Exception {
        // a-管理对象
        Admin admin = connection.getAdmin();
        // b-表不存在时，创建表
        TableName tableName = TableName.valueOf(table);
        if(!admin.tableExists(tableName)){
            // 表描述符-构造器
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
            // 列簇描述符-构造器
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder
                    .newBuilder(Bytes.toBytes(family));
            // 表添加列簇
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
            // 创建表
            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("创建表成功！");
        }
        System.out.println("表存在！");

        // c-关闭
        admin.close();
    }


    @Override
    public void close() throws Exception {
        super.close();
    }
}